package com.doit.day01;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

/**
 * 将kafka的数据，写入到mysql中，并且要求精准一次性消费
 */
public class _01_精准一次性的需求 {
    public static void main(String[] args) throws SQLException {

        //获取到mysql的连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test01", "root", "123456");
        //默认就是开启事务
        conn.setAutoCommit(false);

        PreparedStatement user_info_pps = conn.prepareStatement("insert into user_info values(?,?,?,?)");

        PreparedStatement offset_pps = conn.prepareStatement("insert into tp_offset values(?,?,?,?) on duplicate key update  `offset` = ?");

        PreparedStatement get_offset = conn.prepareStatement("select `offset` from tp_offset where group_id = ? and topic = ? and `partition` = ?");

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //消费者组
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"g001");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //允许自动提交偏移量 __consumer_offset
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
        //kafka消费者的对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
/*

        //订阅主题
//        consumer.subscribe(Arrays.asList("suibianba"));
        TopicPartition tp1 = new TopicPartition("suibianba", 0);
        TopicPartition tp3 = new TopicPartition("suibianba", 2);
        TopicPartition tp2 = new TopicPartition("suibianba", 1);
        consumer.assign(Arrays.asList(tp1,tp2,tp3));

        //指定从哪个topic的哪个分区的哪个偏移量开始读
        //从哪来 ==》 tp_offset
        //指定每一个分区具体是从哪边开始读
        get_offset.setString(1,"g001");
        get_offset.setString(2,"suibianba");
        ResultSet resultSet = get_offset.executeQuery();
        while (resultSet.next()){
            //获取到分区
            int partition = resultSet.getInt(1);
            //获取到偏移量
            long offset = resultSet.getLong(2);
            TopicPartition topicPartition = new TopicPartition("suibianba", partition);
            //指定从哪边开始读
            consumer.seek(topicPartition,offset+1);
        }
*/
        //订阅主题
        consumer.subscribe(Arrays.asList("suibianba"), new ConsumerRebalanceListener() {
            //分配前的策略
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            }

            //分配后的策略
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                //partition存的是什么  我这个消费者现在是消费的那个topic里面的那些分区
                try {
                    for (TopicPartition partition : partitions) {
                        String topic = partition.topic();
                        int partition1 = partition.partition();
                        //通过topic 和 partition 去mysql数据库中找到对应的offset
                        get_offset.setString(1,"g001");
                        get_offset.setString(2,topic);
                        get_offset.setInt(3,partition1);
                        ResultSet resultSet = get_offset.executeQuery();
                        while (resultSet.next()){
                            long offset = resultSet.getLong(1);
                            consumer.seek(partition,offset+1);
                        }
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        });

        while(true){
            //拉取数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            for (ConsumerRecord<String, String> record : poll) {
                try {
                    String topic = record.topic();
                    int partition = record.partition();
                    long offset = record.offset();
                    //这是读过来的数据，需要将它写入到mysql数据库中
                    //1,zss,18,male
                    String value = record.value();
                    String[] arr = value.split(",");
                    Integer id = Integer.parseInt(arr[0]);
                    String name = arr[1];
                    Integer age = Integer.parseInt(arr[2]);
                    String gender = arr[3];

                    if (id == 5){
                        throw new Exception("我自己造的，不服来干我");
                    }

                    //往数据库里面写
                    user_info_pps.setInt(1,id);
                    user_info_pps.setString(2,name);
                    user_info_pps.setInt(3,age);
                    user_info_pps.setString(4,gender);
                    //开始执行sql语句
                    user_info_pps.execute();

                    //提交偏移量
                    offset_pps.setString(1,"g001");
                    offset_pps.setString(2,topic);
                    offset_pps.setInt(3,partition);
                    offset_pps.setLong(4,offset);
                    offset_pps.setLong(5,offset);
                    //执行sql语句
                    offset_pps.execute();

                    //提交事务
                    conn.commit();
                } catch (Exception e) {
                    System.out.println(e);
                    //回滚事务
                    conn.rollback();
                }

            }
        }



    }
}
